package haco.syncremotefile.service;

import cn.hutool.core.io.FileUtil;
import cn.hutool.core.io.IORuntimeException;
import cn.hutool.extra.ftp.Ftp;
import cn.hutool.extra.ftp.FtpException;
import com.aliyun.oss.OSS;
import com.aliyun.oss.model.ListObjectsRequest;
import com.aliyun.oss.model.OSSObjectSummary;
import com.aliyun.oss.model.ObjectListing;
import com.aliyun.oss.model.PutObjectRequest;
import haco.syncremotefile.modal.FTPFileL;
import haco.syncremotefile.modal.STATUS;
import haco.syncremotefile.utils.Download4SourceFtp;
import haco.syncremotefile.utils.PathUtils;
import haco.syncremotefile.utils.SourceConfig;
import haco.syncremotefile.utils.Upload2oss;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.stereotype.Service;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author wangyuwei
 * @history 2021-06-09 09:03
 */
@Slf4j
@Service
@EnableScheduling
public class SyncRemoteFileService {

    @Autowired
    Upload2oss upload2oss;
    @Autowired
    Download4SourceFtp download4SourceFtp ;
    @Autowired
    SourceConfig sourceConfig;
    @Autowired
    SyncLogService syncLogService ;

    @Value("${localTmpPath}")
    String localTmpPath ;

    private static  final  String  SPILT = "/" ;
    private static  final  String  SPILT2 = "\\" ;

    ThreadPoolExecutor executorDownload = new ThreadPoolExecutor(10, 50, 5, TimeUnit.SECONDS,
            new ArrayBlockingQueue(100), new ThreadPoolExecutor.DiscardOldestPolicy());
    ThreadPoolExecutor executorUpload = new ThreadPoolExecutor(10, 50, 5, TimeUnit.SECONDS,
            new ArrayBlockingQueue(100), new ThreadPoolExecutor.DiscardOldestPolicy());

    Lock lock1 = new ReentrantLock();

   // @Scheduled(initialDelay = 10 * 1000 , fixedDelay = 7 * 24 * 60 * 60 * 1000)
    //@Scheduled(cron = "0 0 1  * * MON")   //每周一早上1点开始同步
    public void start(){

        File workspace = new File(PathUtils.get(localTmpPath));
        if(FileUtil.exist(workspace)){
            FileUtil.del(workspace);
        }
        workspace.mkdirs();

        List<SourceConfig.Task> tasks = sourceConfig.getTasks();
        OSS connectOss = upload2oss.getConnectOss();
        for (int i = 0; i <tasks.size(); i++) {
            Ftp ftpConnect = download4SourceFtp.getFtpConnect(tasks.get(i).getSource());
            this.SyncFile(ftpConnect ,connectOss,tasks.get(i),i);
        }
        try {
            Thread.currentThread().join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }


    /**
     * 比较俩文件系统获取差异文件
     * @return
     * @throws Exception
     */
    public void SyncFile (Ftp ftpConnect,OSS ossClient,SourceConfig.Task task,int sourceNum) {
        try {
            SourceConfig.Task.Source source = task.getSource();
            SourceConfig.Task.Target target = task.getTarget();

            List<FTPFileL> allFilePath = download4SourceFtp.getAllFilePath(ftpConnect,
                    source.getDir(), "/",null,sourceNum);
            log.info("获取数据源中的文件完成，其中文件总数:{}" , allFilePath.size());

            ObjectListing objectListing;
            String nextMarker = null;
            List<OSSObjectSummary> sums = null;

            String ossTargetDir = target.getDir();
            if(ossTargetDir.startsWith(SPILT) || ossTargetDir.startsWith(SPILT2)){
                ossTargetDir = ossTargetDir.substring(1);
            }
            if(ossTargetDir.endsWith(SPILT) || ossTargetDir.endsWith(SPILT2)){

            }else{
                ossTargetDir = ossTargetDir + "/";
            }

            do {
                objectListing =
                        ossClient.listObjects(new ListObjectsRequest(upload2oss.getBucketName()).withMarker(nextMarker).withMaxKeys(1000).withPrefix(ossTargetDir));
                if(sums == null){
                    sums = objectListing.getObjectSummaries();
                }else{
                    sums.addAll(objectListing.getObjectSummaries());
                }
                nextMarker = objectListing.getNextMarker();
            } while (objectListing.isTruncated());

            log.info("完成oss文件信息获取...");

            List<FTPFileL> uploadFiles = new ArrayList<>();
            final LinkedBlockingQueue<FTPFileL> queue = new LinkedBlockingQueue<FTPFileL>(1000);

            for (int i = 0; i < allFilePath.size(); i++) {
                FTPFileL ftpFileL = allFilePath.get(i);
                //ftpFileL.setPath(ftpFileL.getPath().replace(source.getDir(),""));

                String fullPath = PathUtils.get(target.getDir(),ftpFileL.getPath() ,ftpFileL.getName());
                boolean flagCompared = false ;
                for (int j = 0; j < sums.size(); j++) {
                    OSSObjectSummary ossObjectSummary = sums.get(j);
                    if (!fullPath.contains(ossObjectSummary.getKey())) {
                        continue;
                    }
                    flagCompared = true ;
                    if(task.getSource().getRelativehour() != 0){
                        ftpFileL.getTimestamp().add(Calendar.HOUR_OF_DAY,task.getSource().getRelativehour());
                    }
                    if (ossObjectSummary.getLastModified().before(ftpFileL.getTimestamp().getTime())) {
                        uploadFiles.add(ftpFileL);
                    } else {
                        break;
                    }
                }
                if(!flagCompared){    //新文件
                    uploadFiles.add(ftpFileL);
                    queue.add(ftpFileL);
                }
            }

            log.info("文件比对完成...");
            log.info("准备工作环境...");
            File workspace = new File(PathUtils.get(localTmpPath,String.valueOf(sourceNum)));
            if( !workspace.exists()){
                workspace.mkdirs();
            }
            //存储下载完成的文件
            BlockingQueue<FTPFileL> filesQueue = new ArrayBlockingQueue<FTPFileL>(1000);

            //存储下载异常的文件
            BlockingQueue<FTPFileL> exceptFiles = new ArrayBlockingQueue<FTPFileL>(1000);

            AtomicInteger downloadFileSize = new AtomicInteger(0);
            log.info("开始文件转储...");
            int uploadFilesSize = uploadFiles.size();
            log.info("待上传文件数：{}",uploadFilesSize);
            new Thread(()-> {
                        int i = 0;
                        while (true) {
                            try {
                                FTPFileL ftpFileL ;
                                if(i >= uploadFilesSize){
                                    ftpFileL = exceptFiles.take();
                                }else{
                                    ftpFileL = uploadFiles.get(i);
                                    i++ ;
                                }

                                int poolSize = executorDownload.getQueue().size();
                                int fileSize = filesQueue.size();
                                while (fileSize > 45 || poolSize > 45 || ftpFileL == null) {
                                    try {
                                        Thread.sleep(3 * 1000);
                                        poolSize = executorDownload.getQueue().size();
                                        fileSize = filesQueue.size();
                                    } catch (InterruptedException e) {
                                        log.error("Thread sleep 异常", e);
                                    }
                                    log.info("下载完成文件池大小：{}", fileSize);
                                    log.info("下载线程池大小：{}", poolSize);
                                }

                                FTPFileL finalFtpFileL = ftpFileL;
                                try {
                                    int finalI = i;
                                    executorDownload.execute(new Runnable() {
                                        @Override
                                        public void run() {
                                            try {
                                                log.info("开始下载文件{}", finalFtpFileL.getAllPath());
                                                ftpConnect.reconnectIfTimeout();
                                                File dir = new File(PathUtils.get(
                                                        localTmpPath,finalFtpFileL.getSourceNumS(), finalFtpFileL.getPath()));
                                                synchronized (SyncRemoteFileService.class) {
                                                    if (dir.exists()) {
                                                        dir.delete();
                                                    }
                                                    dir.mkdirs();
                                                }
                                                File file =
                                                        new File(PathUtils.get(
                                                        localTmpPath, finalFtpFileL.getSourceNumS(),finalFtpFileL.getPath(),finalFtpFileL.getName()));
                                                log.info("准备下载文件{}", LocalDateTime.now());
                                                try {
                                                    OutputStream fileOutputStream = new FileOutputStream(file);
                                                    synchronized (lock1){
                                                        ftpConnect.download(PathUtils.get(source.getDir(), finalFtpFileL.getPath()),
                                                                finalFtpFileL.getName(), fileOutputStream);
                                                        downloadFileSize.incrementAndGet();
                                                        if(downloadFileSize.get() == finalI){
                                                            workspace.delete();
                                                            log.info("删除任务{}本地工作空间",finalFtpFileL.getSourceNumS());
                                                        }
                                                    }
                                                    log.info("完成下载文件：{}", finalFtpFileL.getAllPath());
                                                    log.info("共下载文件数：{}", downloadFileSize);
                                                    fileOutputStream.close();
                                                    filesQueue.add(finalFtpFileL);
                                                } catch (IORuntimeException e) {
                                                    log.info("下载文件阻塞");
                                                    exceptFiles.add(finalFtpFileL);  //下载异常的文件再次加入异常
                                                }
                                            }catch (FtpException e) {
                                                log.error("下载异常，文件追加到下载队列", e);
                                                exceptFiles.add(finalFtpFileL);  //下载异常的文件再次加入异常
                                            } catch (FileNotFoundException e) {
                                                log.error("获取文件{},{}异常", finalFtpFileL.getPath(), finalFtpFileL.getName(), e);
                                            } catch (IOException e) {
                                                log.error("关闭文件流异常", e);
                                            } catch (Exception e) {
                                                log.error("下载文件异常", e);
                                            }
                                        }
                                    });
                                }catch (RejectedExecutionException e){
                                    log.error("executorDownload线程池拒绝,任务重新添加到文件中重新添加到");
                                    exceptFiles.add(finalFtpFileL);  //启动线程异常文件再次加入异常
                                }
                            }catch (InterruptedException e){
                                log.error("获取文件信息异常",e);
                            }
                        }
                    }).start();


            AtomicInteger runningTasks = new AtomicInteger(0);
            new Thread(() -> {
                    while (true) {
                        int poolSizeEx = executorUpload.getQueue().size();
                        while (runningTasks.intValue() > 10) {
                            try {
                                Thread.sleep(3 * 1000);
                            } catch (InterruptedException e) {
                                log.error("Thread sleep 异常", e);
                            }
                        }

                        executorUpload.execute(() -> {
                            runningTasks.incrementAndGet();
                            FTPFileL ftpFileL = null ;
                            try {
                                ftpFileL = filesQueue.take();
                                //ftpFileL = filesQueue.poll(1,TimeUnit.SECONDS);
                                if(ftpFileL == null){
                                    return ;
                                }
                                log.info("开始上传文件:{}", ftpFileL.getAllPath());
                                File file = new File(PathUtils.get(localTmpPath,ftpFileL.getSourceNumS(),
                                        ftpFileL.getPath(), ftpFileL.getName()));
                                PutObjectRequest putObjectRequest = new PutObjectRequest(upload2oss.getBucketName(),
                                        PathUtils.get(target.getDir(), ftpFileL.getPath(), ftpFileL.getName()).substring(1), file);
                                try {
                                    ossClient.putObject(putObjectRequest);
                                    syncLogService.insertLog(task, PathUtils.get(ftpFileL.getPath(), ftpFileL.getName()), STATUS.SUCCESS);
                                } catch (Exception e) {
                                    log.error("上传文件异常{}", ftpFileL.getAllPath(), e);
                                    syncLogService.insertLog(task, PathUtils.get(ftpFileL.getPath(), ftpFileL.getName()), STATUS.FAILED);
                                }
                                file.delete();
                                log.info("文件上传完成{}", ftpFileL.getAllPath());
                            } catch (InterruptedException e) {
                                log.error("队列中获取文件异常", e);
                                e.printStackTrace();
                            } catch (Exception e){
                                log.error("文件上传异常，文件重新放回队列...");
                                if(ftpFileL != null){
                                    filesQueue.add(ftpFileL);
                                }
                            }finally {
                                runningTasks.decrementAndGet();
                            }
                        });
                    }
                }).start();
        } catch (Exception e){
            log.error("SyncFile异常",e);
        }finally {
           /* try {
                log.info("关闭ftp连接");
                ftpConnect.close();
            } catch (IOException e) {
                e.printStackTrace();
            }*/
        }
    }






}
